package com.atguigu.canal.utils;

import org.apache.kafka.clients.KafkaClient;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.io.IOException;
import java.util.Properties;

/**
 * kafka工具类
 *
 * @author WangJX
 * @version 1.0
 * @date 2019/11/28 10:00
 */
public class MyKafkaSender {

    private static KafkaProducer<String, String> kafkaProducer;

    //获取kafka producer连接
    public static KafkaProducer<String, String> createKafkaProducer() throws IOException {
        //读取kafka配置文件
        Properties pro = new Properties();
        pro.load(MyKafkaSender.class.getClassLoader().getResourceAsStream("kafka.properties"));

        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(pro);
        return kafkaProducer;
    }


    //发送数据到kafka
    public static void send(String topic, String value) {
        try {
            if (kafkaProducer == null) {
                kafkaProducer = createKafkaProducer();
            }

            //发送数据
            kafkaProducer.send(new ProducerRecord<String, String>(topic, value));

        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
